package com.xiaoniu.tcp.server;

import com.xiaoniu.MsgUtils;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Vertx;
import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;
import io.vertx.core.net.NetServer;
import io.vertx.core.net.NetServerOptions;
import io.vertx.core.net.NetSocket;
import io.vertx.core.parsetools.RecordParser;
import io.vertx.core.streams.ReadStream;

import java.util.*;

/**
 * Created by sweet on 2018/1/2.
 * ---------------------------
 */
public class TcpServerVerticle extends AbstractVerticle {

    public static void main(String[] args) {
        System.setProperty("vertx.logger-delegate-factory-class-name",
                "io.vertx.core.logging.Log4j2LogDelegateFactory");
        Vertx.vertx().deployVerticle(new TcpServerVerticle());
    }

    private final Logger LOGGER = LoggerFactory.getLogger(TcpServerVerticle.class);

    private static final Map<String, NetSocket> SOCKET_MAP = new HashMap<>();
    private static final Map<String, Long> ACTIVE_SOCKET_MAP = new HashMap<>();

    @Override
    public void start() throws Exception {
        NetServerOptions options = new NetServerOptions();
        options.setTcpKeepAlive(true);

        final RecordParser parser = RecordParser.newDelimited("\n", h -> {
            final String msg = h.toString();
            LOGGER.debug("服务器解包: " + msg);

            final String[] msgSplit = msg.split("\\*");
            final String socketId = msgSplit[0];
            final String msgBody = msgSplit[1];

            if ("PING".equals(msgBody)) { // 心跳
                ACTIVE_SOCKET_MAP.put(socketId, System.currentTimeMillis());
                SOCKET_MAP.get(socketId).write(MsgUtils.joinMsg(socketId, "PING"));
            } else {
                // 其他信息，这里简单模拟一下，原样返回给客户端
                SOCKET_MAP.get(socketId).write(MsgUtils.joinMsg(socketId, msgBody));
            }
        });

        NetServer server = vertx.createNetServer(options);

        ReadStream<NetSocket> stream = server.connectStream();
        stream.handler(netSocket -> {
            String socketId = netSocket.writeHandlerID();
            LOGGER.debug("New socket Id: " + socketId);
            netSocket.handler(parser::handle);
            netSocket.write(socketId + "*" + "Server\n");

            SOCKET_MAP.put(socketId, netSocket);
            ACTIVE_SOCKET_MAP.put(socketId, System.currentTimeMillis());
        });

        stream.endHandler(end -> {
            System.out.println("stream end");
        });

        stream.exceptionHandler(ex -> {
            System.out.println("stream ex : " + ex.getMessage());
            ex.printStackTrace();
        });

        // 检查心跳
        vertx.setPeriodic(1000l * 60, t -> {
            System.out.println("SOCKET MAP");
            System.out.println(SOCKET_MAP);
            System.out.println("ACTIVE MAP");
            System.out.println(ACTIVE_SOCKET_MAP);

            Iterator<Map.Entry<String, Long>> iterator = ACTIVE_SOCKET_MAP.entrySet().iterator();
            while (iterator.hasNext()) {
                Map.Entry<String, Long> entry = iterator.next();
                long time = System.currentTimeMillis() - entry.getValue();
                if (time > (1000 * 60)) {
                    LOGGER.debug("SocketId: " + entry.getKey() + ", 被清除");
                    SOCKET_MAP.remove(entry.getKey()).close();
                    iterator.remove();
                }
            }
        });

        server.listen(8080, res -> {
            if (res.succeeded()) {
                System.out.println("Server start !!!");
                LOGGER.debug("Tcp Server Start");
            } else {
                res.cause().printStackTrace();
            }
        });

    }
}
